package com.zhangdi.mgsync.core.listener;

import com.mongodb.client.model.changestream.ChangeStreamDocument;
import com.zhangdi.mgsync.core.adapter.MyDocument;
import com.zhangdi.mgsync.core.adapter.SyncAdapter;
import com.zhangdi.mgsync.core.adapter.SyncFactory;
import com.zhangdi.mgsync.core.annotation.SyncFlag;
import com.zhangdi.mgsync.core.cache.Cache;
import com.zhangdi.mgsync.core.cache.CacheManager;
import com.zhangdi.mgsync.core.common.MgSyncConst;
import com.zhangdi.mgsync.core.utils.ReflectionUtils;
import org.bson.Document;
import org.springframework.data.mongodb.core.messaging.Message;
import org.springframework.data.mongodb.core.messaging.MessageListener;

import java.util.HashMap;
import java.util.Map;
import java.util.Set;

/**
 * Created by zhangdi on 2020/07/15 16:19:40.
 */
public class TopicMessageListener implements MessageListener<ChangeStreamDocument<Document>, Object>, ListenerCacheManager {
    private String scanAdapterPackage;
    private SyncFactory<MyDocument> factory;
    private Map<String, SyncAdapter<MyDocument>> syncApiMap = new HashMap<>();
    private CacheManager cacheManager;
    private Cache<String, String> token_cache;

    public TopicMessageListener(String scanAdapterPackage, SyncFactory<MyDocument> factory) {
        this.scanAdapterPackage = scanAdapterPackage;
        this.factory = factory;
        this.init();
    }

    private void init() {
        Set<Class<?>> apis = ReflectionUtils.queryClassBySyncFlag(scanAdapterPackage);
        for (Class<?> clazz : apis) {
            SyncAdapter<MyDocument> api = null;
            try {
                //noinspection unchecked
                api = (SyncAdapter<MyDocument>) clazz.newInstance();
            } catch (InstantiationException | IllegalAccessException e) {
                e.printStackTrace();
            }
            String collection = clazz.getAnnotation(SyncFlag.class).collection();
            syncApiMap.put(collection, api);
        }
    }

    @Override
    public void onMessage(Message<ChangeStreamDocument<Document>, Object> message) {
        MyDocument myDocument = (MyDocument) message.getBody();
        if (null == myDocument) {
            //处理异常
            return;
        }

        getCacheManager().put(MgSyncConst.KEY, message.getRaw().getResumeToken().toJson());

        if (syncApiMap == null || syncApiMap.size() == 0) {
            throw new RuntimeException("call init method first!");
        }

        String collectionName = myDocument.getClass().getSimpleName().toLowerCase();
        if (!syncApiMap.containsKey(collectionName)) {
            return;
        }

        if (message.getRaw() == null) {
            return;
        }

        factory.execute(syncApiMap.get(collectionName), message.getRaw(), myDocument);
    }

    @Override
    public Cache<String, String> getCacheManager() {
        if (null == token_cache) {
            token_cache = cacheManager.getCache(MgSyncConst.CACHE_NAME);
        }
        return token_cache;
    }

    @Override
    public void setCacheManager(CacheManager cacheManager) {
        this.cacheManager = cacheManager;
    }
}
